package cn.doitedu.rtmk.demo8;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
{
 "rule_id":"rule-2-1",
 "event_id":"e"
}
 */
public class RuleModel2Calculator implements RuleCalculator{

    Roaring64Bitmap preSelectUsersBitmap;
    String eventId;
    String ruleId;
    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr, Roaring64Bitmap preSelectUsersBitmap) throws IOException {
        this.preSelectUsersBitmap = preSelectUsersBitmap;
        JSONObject jsonObject = JSON.parseObject(ruleParamJsonStr);
        eventId = jsonObject.getString("event_id");
        ruleId = jsonObject.getString("rule_id");

        message = new JSONObject();
    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        if(eventId.equals(eventBean.getEventId())){
            message.put("rule_id",ruleId);
            message.put("user_id",eventBean.getUid());
            message.put("timestamp",eventBean.getTimestamp());
            collector.collect(message.toJSONString());
        }


    }
}
